package vip.xls8z.reactor.jdk9.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

class MyProcessor extends SubmissionPublisher<Integer> implements Flow.Processor<Integer, Integer> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("onSubscribe - 收到订阅请求");
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("onNext - 收到发布者数据: " + item);
        if (item % 2 == 0) {
            this.submit(item);
        }
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        System.out.println("处理器处理完毕.");
        this.close();
    }
}

public class FlowDemo2 {
    public static void main(String[] args) throws InterruptedException {
        // 1. 定义发布者，发布的数据类型是Integer
        // 直接使用JDK自带的SubmissionPublisher，它实现了Publisher接口
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        MyProcessor processor = new MyProcessor();

        // 定义订阅者
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription = null;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("onNext 从Processor 接受到过滤后的数据：" + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
                subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete 所有数据接收完成");
            }
        };
        // 发布者和处理器建立关系，此时处理器扮演订阅者
        publisher.subscribe(processor);

        // 处理器和订阅者建立关系，此时处理器扮演发布者
        processor.subscribe(subscriber);

        // 生产数据并发布
        for (int i = 0; i < 10; i++) {
            publisher.submit(i);
            Thread.sleep(1000);
        }

        // 数据处理完，关闭publisher
        publisher.close();

        // 主线程延迟停止
        Thread.sleep(1000);
    }
}
